package com.wlz.rocketmq.transactionmessage;

import com.wlz.rocketmq.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.*;

/**
 *  事务消息 -- 生产者
 *
 * @author wlz
 * @date 2022-07-02  4:09 下午
 */
@Slf4j
public class TransactionProducer {

    public static void main(String[] args) throws Exception{
        TransactionListener transactionListener = new TransactionListenerImpl();

        TransactionMQProducer producer = new TransactionMQProducer("TransactionProducer");

        producer.setNamesrvAddr(Constants.NAMESVADDR);
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        producer.setExecutorService(executorService);

        producer.setTransactionListener(transactionListener);

        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};

        for (int i = 0; i < 10; i++) {

            try {
                Message msg = new Message("TransactionTopic", tags[i % tags.length],
                        "KEY" + i, ("hello world" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

                TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(msg, null);

                log.info("TransactionSendResult : {}", transactionSendResult);

                Thread.sleep(10);
            } catch (Exception e) {
                log.error("TransactionProducer error: {}", e);
            }
        }

        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();



    }
}
